package com.oct.ywsc.job.kafka;

import com.alibaba.fastjson2.JSONObject;
import com.oct.ywsc.core.bean.MsgBean;
import com.oct.ywsc.core.config.KafkaConsumerConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;


import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.kafka.core.ConsumerFactory;

/**
 * kafka消息
 */
@Slf4j
public class MQClient implements IClient {

    private KafkaConsumer<String, String> consumer;    // kafka消费端

    private Map<String, Object> customerProps;


    private List<String> topics;


    public MQClient(Map<String, Object> customerProps, List<String> topics) {
        this.customerProps = customerProps;
        this.topics = topics;
    }

    @Override
    public void init() {
        //  创建 Kafka 消费者
        consumer = new KafkaConsumer<>(customerProps);
        // 订阅主题
        consumer.subscribe(topics);
        log.info(String.format("订阅topic:[%s]", JSONObject.toJSONString(topics)));
    }

    @Override
    public List<MsgBean> poll(Duration maxPollIntervalMs) {

        List<MsgBean> list = null;
        ConsumerRecords<String, String> records = consumer.poll(maxPollIntervalMs);
        if (records!= null && records.count() > 0) {
            list = new ArrayList<>();
            StringBuilder sb = new StringBuilder();
            for (ConsumerRecord<String, String> record : records) {
                sb.append(String.format("offset: [%s] value:[%s]  \n",record.offset(), record.value()));
                MsgBean mb = new MsgBean(record.offset(), record.value());
                list.add(mb);
            }
            log.info(String.format("从kafka消费端获取到消息:  \n%s \n 共%s条",sb.toString(), list.size()));
        }


        return list;
    }

    @Override
    public void close() {
        if (consumer != null){
            consumer.close();
        }
    }

    @Override
    public void commitSync() {
        this.consumer.commitSync();
    }


}
